package com.bw.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.text.SimpleDateFormat;
import java.util.Date;

public class UniqueVisitAPP {
    // 数据源  App--->nginx--->springboot---kafka(ods_base_log)--->flink(dwd)--->kafka(dwd)--->flink(dwm)---->kafka(dwm)
    // 程序
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //        开启CK，创建文件夹
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:9820/gmall-flink/ck"));
//         每隔5秒保存一次
//        env.enableCheckpointing(5000L);
//        精准消费一次
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        配置超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(100000L);
//        同时最大可以运行几个ck
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        上一个ck的尾部到下一个ck的头之间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2. 读取Kafka dwd_page_log
        String topic = "dwd_page_log";
        String groupId = "unique_visit_app_2105b";
        String sinkTopic = "dwm_unique_visit";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, groupId));
        // 3. 把每一行数据变成Json数据
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String value) throws Exception {
                return JSON.parseObject(value);
            }
        });
        // 4. 过滤数据，只要每天每个Key第一条数据
        KeyedStream<JSONObject, String> keyedStream = jsonObjDS.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        });

        SingleOutputStreamOperator<JSONObject> uvDS = keyedStream.filter(new RichFilterFunction<JSONObject>() {
            private ValueState<String> valueState;
            private SimpleDateFormat simpleDateFormat;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("value-date", String.class);
                StateTtlConfig ttl = new StateTtlConfig.Builder(Time.hours(24))
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)// 当创建和写入的时候，重新开启倒计时
                        .build();
                valueStateDescriptor.enableTimeToLive(ttl);// 状态24时过期
                /**
                 *  mid   last_page_id  日期            valueState
                 *
                 *   1      null       2020-02-02 8:00       2020-02-02    24过期
                 *   1      null       2020-02-03 7:00       2020-02-03    距离过期还剩1小时
                 *                                                        2020-02-03 8:00状态清空
                 *   1      null       2020-02-03 9:00       2020-02-03
                 */


                // 初始化状态
                valueState = getRuntimeContext().getState(valueStateDescriptor);
                simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
            }

            @Override
            public boolean filter(JSONObject jsonObject) throws Exception {
                /**
                 *  mid   last_page_id  日期
                 *  1       null        2024-02-02:8:00
                 *  1       detail      2024-02-02:9:00
                 *  1       null       2024-02-02:10:00
                 *  1       null       2024-02-03:10:00
                 *  1       detail     2024-02-04:10:00
                 *
                 *  1       null      2024-02-24 23:59:59
                 *  1       detail    2024-02-25 00:00:01
                 *
                 *
                 *  1       null     2024-02-26 8:00
                 *  1       null     2024-04-26 8:00
                 */
                String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                // 找出上一条为null的数据
                if (lastPageId == null || lastPageId.length() <= 0) {
                    String lastDate = valueState.value();
                    String currentDate = simpleDateFormat.format(new Date(jsonObject.getLong("ts")));
                    // 访问日期跟状态日期不等
                    if (!currentDate.equals(lastDate)) {
                        valueState.update(currentDate);// 2020-02-02
                        return true;
                    }
//                    else{
//                        return false;
//                    }
                }
                return false;
            }
        });
        // 5. 打印数据，并保存到dwm主题当中
        uvDS.print();
        uvDS.map(JSONAware::toJSONString).addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));
        // 6. 启动任务
        env.execute("UniqueVisitAPP");
    }
}
